package dnet

import (
	"dgo/goutils/utils"
	"fmt"
	"math"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

type TcpServerV0 struct {
	sendCnt0   int32
	sendCnt1   int32
	id         string
	dna        uint32
	flag       byte
	addr       string
	LogTypeStr string
	closeW     sync.WaitGroup
	netListen  net.Listener

	isTerminated            bool
	startFlag               int32
	BlockRecvSize           uint16
	timeout_sec_kick        int // 如果>0， 则进行心跳检测
	checkTimeoutTaskCnt     int32
	checkTimeoutIntervalDNA int
	checkTimeoutInterval    time.Duration       // 检测超时任务间隔
	maxOnline               uint32              // 最大连接数, 超过最大连接数据会随机删除一个连接， 0：不限制
	onlineConnections       utils.GSyncMapGroup // 在线列表
	onlineCount             uint32              // 在线数量
	MonitorFlag             int                 // 监控类型 0:不跟踪, 1:监控每个连接详细
	UseQueueSend            bool                // 使用队列进行发送, 如果是队列. 必须保证Buf 不能被重复修改, 队列发送, 是异步进行的, 默认为false
	OnRecvBufEvent          func(sender *TcpServerClientConnectionV0, buf []byte)
	OnDisEvent              func(sender *TcpServerClientConnectionV0)
	OnConnEvent             func(sender *TcpServerClientConnectionV0)
	OnErr                   func(sender *TcpServerV0, err error)
	OnConnectionErr         func(sender *TcpServerClientConnectionV0, err error)
}

func NewTcpServerV0(id string, addr string, maxCount int32, timeout_sec int) *TcpServerV0 {
	rval := &TcpServerV0{
		dna:              utils.NewObjectDNA(),
		flag:             0,
		startFlag:        0,
		id:               id,
		addr:             addr,
		BlockRecvSize:    RECV_BLOCK_SIZE,
		maxOnline:        uint32(maxCount),
		timeout_sec_kick: timeout_sec,
	}
	return rval
}

func NewTcpServerV0Ex(id string) *TcpServerV0 {
	rval := &TcpServerV0{
		dna:              utils.NewObjectDNA(),
		flag:             0,
		startFlag:        0,
		id:               id,
		BlockRecvSize:    RECV_BLOCK_SIZE,
		maxOnline:        0,
		timeout_sec_kick: 0,
	}
	return rval
}

func (this *TcpServerV0) SetAddress(addr string) bool {
	if this.addr != addr {
		this.addr = addr
		if this.startFlag == 1 {
			this.Stop()
			this.Start()
		}
		return true
	}
	return false
}

func (this *TcpServerV0) SetMaxOnline(cnt uint32) bool {
	if this.maxOnline != cnt {
		this.maxOnline = cnt
		return true
	}
	return false
}

func (this *TcpServerV0) GetActive() bool {
	return this.startFlag == 1
}

func (this *TcpServerV0) GetOnlineCnt() uint32 {
	return this.onlineCount
}

func (this *TcpServerV0) incSendCnt0() {
	atomic.AddInt32(&this.sendCnt0, 1)
}

func (this *TcpServerV0) incSendCnt1() {
	atomic.AddInt32(&this.sendCnt1, 1)
}

func (this *TcpServerV0) GetSendStatus() string {
	return fmt.Sprintf("%d/%d r:%d", this.sendCnt1, this.sendCnt0, this.sendCnt0-this.sendCnt1)
}

func (this *TcpServerV0) GetMaxOnline() uint32 {
	return this.maxOnline
}

func (this *TcpServerV0) GetAddr() string {
	return this.addr
}

func (this *TcpServerV0) ConfigKickTimeOut(val int) bool {
	if this.timeout_sec_kick != val {
		this.timeout_sec_kick = val
		if this.timeout_sec_kick > 0 && this.checkTimeoutInterval == 0 {
			this.checkTimeoutInterval = time.Duration(math.Floor(float64(this.timeout_sec_kick)/2)) * time.Second

			if this.startFlag == 1 { // 已经开启了服务
				this.restartKickTask()
			}
		}

		return true
	}
	return false
}

func (this *TcpServerV0) ConfigCheckTimeoutInterval(secs int) bool {
	if int(this.checkTimeoutInterval.Seconds()) != secs {
		this.checkTimeoutIntervalDNA++
		this.checkTimeoutInterval = time.Duration(secs) * time.Second
		if this.startFlag == 1 { // 已经开启了服务
			this.restartKickTask()
		}
		return true
	}
	return false
}

func (this *TcpServerV0) restartKickTask() {
	if this.checkTimeoutInterval >= time.Second {
		dna := this.checkTimeoutIntervalDNA
		this.checkTimeoutTaskCnt++
		utils.AddTaskFxiedInterval(fmt.Sprintf("tcpsvr_%d", this.dna), this.checkTimeoutInterval, func() bool {
			r := dna == this.checkTimeoutIntervalDNA
			if !r {
				this.checkTimeoutTaskCnt--
				return false
			}
			if this.timeout_sec_kick > 0 {
				this.onlineConnections.Range(func(key, value interface{}) bool {
					ctx := value.(*TcpServerClientConnectionV0)
					ctx.checkKickOut(this.timeout_sec_kick)
					return true
				})
			}
			return true
		})
	}
}

func (this *TcpServerV0) RangeOnlineList(fn func(conn *TcpServerClientConnectionV0) bool) (cnt int) {
	// 踢掉较早的连接
	onlines := this.onlineConnections.ListSorted(func(iItm, jItm interface{}) bool {
		iConn := iItm.(*TcpServerClientConnectionV0)
		jConn := jItm.(*TcpServerClientConnectionV0)
		return iConn.handle < jConn.handle
	})

	for i := 0; i < len(onlines); i++ {
		if !fn(onlines[i].(*TcpServerClientConnectionV0)) {
			break
		}
		cnt++
	}
	return
}

func (this *TcpServerV0) innerCheckKickOut() {
	if this.maxOnline == 0 {
		return
	}

	if this.maxOnline > this.onlineCount {
		return
	}

	// 踢掉较早的连接
	onlines := this.onlineConnections.ListSorted(func(iItm, jItm interface{}) bool {
		iConn := iItm.(*TcpServerClientConnectionV0)
		jConn := jItm.(*TcpServerClientConnectionV0)
		return iConn.connecttime.Before(jConn.connecttime)
	})

	cnt := len(onlines)
	for {
		for i := 0; i < len(onlines); i++ {
			if cnt < int(this.maxOnline) {
				break
			}

			conn := onlines[i].(*TcpServerClientConnectionV0)
			conn.RequestDisconnect("新的连接进入, 请求断开先接入的连接")
			cnt--
		}
	}
}

func (this *TcpServerV0) innerStartListen() {
	utils.Log.InfoTypef(this.LogTypeStr, "%s侦听地址:%s", this.id, this.addr)
	netListen, err := net.Listen("tcp", this.addr)
	if err != nil {
		if this.OnErr != nil {
			this.OnErr(this, err)
		}
		return
	}

	this.netListen = netListen
	defer netListen.Close()

	this.isTerminated = false
	this.closeW.Add(1)
	defer this.closeW.Done()
	for {
		_conn, err := netListen.Accept()
		if err != nil {
			if this.isTerminated {
				break
			} else {
				if this.OnErr != nil {
					this.OnErr(this, err)
				}
				time.Sleep(time.Second * 10)
				continue
			}
		} else {
			// 检测是否要T掉之前的连接
			this.innerCheckKickOut()
		}

		if this.isTerminated {
			_conn.Close()
			break
		}
		ctx := NewCommTcpSvrV0ClientConnection(_conn, this)
		ctx.connecttime = time.Now()
		this.addOnline(ctx)
		go ctx.Start()
	}
}

/*
*

	根据Handle查找连接(在线)
*/
func (self *TcpServerV0) FindConnByHandle(handle uint32) (ctx *TcpServerClientConnectionV0) {
	obj := self.onlineConnections.Get(handle, nil)
	if obj == nil {
		return nil
	}
	return obj.(*TcpServerClientConnectionV0)
}

func (self *TcpServerV0) RequestDisByHandle(handle uint32, disReson string) bool {
	ctx := self.FindConnByHandle(handle)
	if ctx != nil {
		ctx.RequestDisconnect(disReson)
		return true
	}
	return false
}

func (self *TcpServerV0) addOnline(ctx *TcpServerClientConnectionV0) {
	_, loaded := self.onlineConnections.LoadOrStore(ctx.handle, ctx)
	if !loaded {
		atomic.AddUint32(&self.onlineCount, 1)
	} else {
		utils.Log.WarnTypef(self.LogTypeStr, "连接(%d:%s)进入, handle重复!, 当前在线数:%d", ctx.handle, ctx.remoteAddr, self.onlineCount)
	}
	if self.OnConnEvent != nil {
		self.OnConnEvent(ctx)
	}
}

func (self *TcpServerV0) removeOnline(ctx *TcpServerClientConnectionV0) {
	if self.onlineConnections.Contain(ctx.handle) {
		utils.AtomicDecUint32(&self.onlineCount, 1)
	} else {
		utils.Log.WarnTypef(self.LogTypeStr, "连接(%d:%s)断开, 当前在线:%d, Onlines不包含Handle", ctx.handle, ctx.remoteAddr, self.onlineCount)
	}
	self.onlineConnections.Remove(ctx.handle)
	ctx.distime = time.Now()

	if ctx.OnDisEvent != nil {
		ctx.OnDisEvent(ctx)
	}

	if self.OnDisEvent != nil {
		self.OnDisEvent(ctx)
	}
}

func (this *TcpServerV0) onRecvBuffer(ctx *TcpServerClientConnectionV0, buf []byte) {
	if this.OnRecvBufEvent != nil {
		this.OnRecvBufEvent(ctx, buf)
	}
}

func (this *TcpServerV0) DisconnectAll() {
	this.onlineConnections.Range(func(key, value interface{}) bool {
		ctx := value.(*TcpServerClientConnectionV0)
		ctx.RequestDisconnect("主动请求断线(DisconnectAll)")
		return true
	})
}

func (this *TcpServerV0) Start() {
	if !atomic.CompareAndSwapInt32(&this.startFlag, 0, 1) {
		return
	}

	this.isTerminated = false
	if this.timeout_sec_kick > 0 {
		this.restartKickTask()
	}
	utils.GoFunc2(this.innerStartListen)
}

/*
**

	会阻塞
*/
func (this *TcpServerV0) StartSync() {
	if !atomic.CompareAndSwapInt32(&this.startFlag, 0, 1) {
		return
	}

	this.isTerminated = false
	if this.timeout_sec_kick > 0 {
		utils.GoFunc2(func() {
			this.restartKickTask()
		})
	}
	this.innerStartListen()
}

func (this *TcpServerV0) Stop() {
	if !atomic.CompareAndSwapInt32(&this.startFlag, 1, 0) {
		return
	}
	this.checkTimeoutIntervalDNA++
	this.checkTimeoutTaskCnt-- // DNA改变会导致任务停止
	this.isTerminated = true
	this.netListen.Close()
	this.DisconnectAll()
	this.closeW.Wait()
}

func (this *TcpServerV0) Send2All(buf []byte) int {
	defer utils.PanicHandler()
	cnt := 0
	this.onlineConnections.Range(func(key, value interface{}) bool {
		ctx := value.(*TcpServerClientConnectionV0)
		err := ctx.Send(buf)
		if err != nil {
			cnt++
		}
		return true
	})

	return cnt
}
